IO模型

1. IO复用

进程需要这样一种功能：内核一旦发现进程指定的一个或多个IO条件（事件）就绪（输入准备好被读取或者输出准备好被输出），它通知进程，这个就叫做I/O复用。

IO复用的典型使用场景

• 客户处理多个描述符号（通常是交互式输入和网络套接字）时候，必须使用I/O复用，例如实际的网络聊天，进程既要等待用户的输入，又要处理来自网络套接字的输入，采用I/O复用可以大幅度提高并发度和节省资源；

• TCP服务器既要处理监听套接字，又要处理已经连接的套接字，这时候需要I/O复用以提高并发性能；实际上常见的并发连接的客户数多，但是每个连接的任务轻（CPU时间短）比如多人在线聊天室；连接并发数多，但是每个连接不都是长期处于活跃状态，可能是偶尔I/O一些数据，这时候需要I/O复用，让系统别被这些Lazy的客户端拖垮；

• 如果一个服务器需要同时处理多个服务或协议，一般使用IO复用；

其实，不难看出I/O复用的目的是为了让计算机的资源得到尽可能最大化的利用，提高并行化程度达到这种目的的一种有效途径。

I/O复用在某种程度上和操作系统的中断处理机制何其的相似。

操作系统作为计算机的核心软件，管理着众多的硬件设备，具有极大的并发性能，我们可以同时听音乐，同时拷文件，同时跑程序等等；如此高的并发能力得益与中断机制给操作系统带来的异步能力，当某个硬件设备特定的事件发生时（比如磁盘就绪），就引发一个硬件中断，操作系统能捕获到这些中断，将当前运行的进程挂起，然后转而执行该中断对应的中断处理程序。

这种机制有点贪心的思想，操作系统总是尽可能的少等待，有活就干，没活就离开；这种机制中，众多硬件设备就类似与IO复用里面的众多Lazy的客户，他们需要与服务器（这里是操作系统）保持连接（做到随叫随到），但是又总是需要操作系统。中断是操作系统运行的动力源泉，这是一个典型的异步模型。I/O复用依赖于内核提供的异步能力，其抽象程度与操作系统和硬件设备的模型相似，这样能充分利用计算机的资源（软件充分利用操作系统，操作系统充分利用硬件）。

2. I/O模型

• 阻塞IO
• 非阻塞IO
• IO复用
• 信号驱动IO
• 异步IO

应用程序典型的I/O包含两个不同的阶段：

以输入为例

1. IO请求就绪阶段 应用程序向内核发出请求，检查硬件设备是否准备好数据（例如等待数据从网络到达、用户键盘敲击）；若硬件设备就绪，产生中断；
2. IO实际读写阶段 内核捕获处理中断，然后内核空间将数据拷贝到用户的进程空间中，CPU切换到用户态；

这两个阶段的产生主要是由于计算机目前的体系结构决定的（最顶层用户程序运行在用户态，中间层内核工作在内核态，最底层是硬件）；这两个阶段都有可能产生阻塞，在内核与硬件之间，应付这种阻塞采用了基于事件的中断机制以提高并发性；而在内核与用户进程之间，异步IO机制也将为用户进程并发程度提高带来曙光。

2.1 阻塞IO

最普遍的IO模型都是阻塞模型，这种模型在IO的的第一阶段（向内核发出I/O请求的系统调用）时，被阻塞直到I/O就绪，再执行实际I/O操作；

2.2 非阻塞IO

进程通知内核，进程所发起的IO请求在I/O未就绪的条件下，也不能把进程阻塞（挂起），而是返回一个错误；该模型可能需要进程不断的去轮询某个FD是否IO就绪，这样的轮询模型比较少见，轮询会额外的浪费大量的CPU资源。

2.3 IO复用

该模型本质上属于非阻塞模型I/O，常见的是select和poll机制。对于普通的非阻塞模型，一个进程或者线程始终轮询同一个FD，这样效率较低，特别是对于同时需要处理很多FD的程序；

而IO复用模型，可以让进程或者线程同时管理（轮询）多个FD，以select为例，select能够管理许多FD上的不同事件，select调用在所有FD都没有IO事件(就绪、可读、可写等)产生时会阻塞，虽然select调用会阻塞，但是其管理轮询的FD不是阻塞的，因此该模型依然算是非阻塞IO模型；

该模型通常和阻塞IO模型+多线程作对比，在典型的socket服务器端，该模型和阻塞IO模型+多线程可以实现相同的功能，大多数情况下效果性能也不会差太离谱；

但是考虑在某些特定的场景下，例如客户端连接数目众多，但是每个连接今次进行IO次数比较多，IO的数据比较少（典型的就是多人在线聊天，大量的客户端同服务器保持连接，但是每一个连接“走走停停”，隔一会来几个字节的数据），在阻塞IO模型+多线程模型中，每一个连接对应于一个线程，大量的长连接会很快消耗掉计算机中（也可能是线程池）中所有线程资源，从而拒绝了额外的新的客户连接请求，但是此时计算机的CPU却是闲置的，因为很多线程其实是在阻塞状态，因此这种场景下，该模型没有很好的提高系统的并发性；

而采用I/O复用模型，用一个或多个线程去select或者poll检查多个连接上的IO事件，当IO事件(可读、可写等)发生，就利用新开线程池中的线程来处理IO，由于每一次的IO的任务很轻，并且IO数据已经来到了内核或者进程的缓冲区，这样新开的线程很快就会被线程池回收，这样当新来的客户连接请求进来，系统仍然有足够的线程资源来处理请求，极大的提高了该场景下系统的吞吐量，IO复用模型要远优于阻塞IO模型+多线程；

实际应用中，辨证的选择IO复用模型和阻塞IO模型+多线程。

2.4 信号驱动IO

进程不会阻塞在第一阶段，当IO就绪时，内核会通过信号的形式通知进程；

2.5 异步IO

进程在整个IO过程（IO请求和IO读写阶段）都不会阻塞，而是在IO完成后由内核通知进程；

2.6 同步IO与异步IO

在某些术语中，阻塞和非阻塞用于区分IO的第一个阶段，即IO的请求阶段；
同步和异步用于区分IO的第二个阶段，即IO的读写阶段；

因此，按照这个来重新分类上述五种IO模型，就可以分为

• 同步IO

  • 阻塞IO (最普遍的同步阻塞BIO)
  • 非阻塞IO （Java NIO， 需要轮询，不停的轮询某个FD是否就绪）
  • IO复用 (Java NIO的Select机制，同时监听多个FD，虽然单个FD不阻塞，但是当所有FD都没有就绪时，select调用会阻塞；一个阻塞换来多个不阻塞)
  • 信号驱动IO (同同步非阻塞IO，不用轮询，而是直接接收内核发来的IO信号)

• 异步IO （Java AIO， IO的两个阶段都不会阻塞，进程只需要发起IO请求，在IO完成两个阶段后，由内核负责通知进程，期间进程都不会挂起）

2.7 5种IO模型的Java示例

• （1）阻塞IO + 多线程

package iomodel;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.Reader;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

public class BIODemo {

	public static void main(String[] args) {
		// TODO Auto-generated method stub
		Executor pools = Executors.newFixedThreadPool(5);
		try {
			ServerSocket sS = new ServerSocket(8888);
			for (;;) {
				Socket s = sS.accept();//blocking
				pools.execute(new BIOWorker(s));
			}
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

}

class BIOWorker implements Runnable {
	Socket client = null;

	public BIOWorker(Socket client) {
		super();
		this.client = client;
	}

	@Override
	public void run() {
		// TODO Auto-generated method stub
		try {
			BufferedReader is = new BufferedReader(new InputStreamReader(client.getInputStream()));
			String line = null;
			while ((line = is.readLine()) != null)//blocking
				System.out.println(line);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	
}

• （2）非阻塞IO模型

package iomodel;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.nio.ByteBuffer;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;
import java.nio.charset.CharsetDecoder;
public class NBIODemo {

	public static void main(String[] args) {
		// TODO Auto-generated method stub
		try {
			Charset ascii = Charset.forName("us-ascii");
			CharsetDecoder decoder = ascii.newDecoder();
			ServerSocketChannel ssc = ServerSocketChannel.open();
			ssc.configureBlocking(false);
			ServerSocket ss = ssc.socket();
			ss.bind(new InetSocketAddress("localhost", 8888));

			SocketChannel sC = null;

			while ((sC = ssc.accept()) == null)
				System.out.println("i am pooling for new connection!");// poll until new connection was established.non-blocking
			sC.configureBlocking(false);

			ByteBuffer buffer = ByteBuffer.allocate(1024);
		
			while (true) {//pool for reading
				int r = sC.read(buffer);//non-blocking
				System.out.println("i am pooling for new data!");
				if (r > 0) {
					buffer.flip();
					System.err.println("recv " + r +" bytes from " + sC);
					System.err.print(decoder.decode(buffer));
					buffer.compact();
				}
			}

		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

}

• （3）I/O复用

package iomodel;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;
import java.nio.charset.CharsetDecoder;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

public class MultiplexingIODemo {

	public static void main(String[] args) {
		// TODO Auto-generated method stub
		Executor pool = Executors.newFixedThreadPool(5);
		try {
			Selector selector = Selector.open();
			ServerSocketChannel ssc = ServerSocketChannel.open();
			ssc.configureBlocking(false);// listening socket channel is
											// non-blocking for accept()

			ServerSocket ss = ssc.socket();
			ss.bind(new InetSocketAddress("localhost", 8888));

			ssc.register(selector, SelectionKey.OP_ACCEPT);

			for (;;) {

				// System.out.println("I am selecting..");
				selector.select();// may block until there occurred at least one
									// IO event.

				Set<SelectionKey> selectKeys = selector.selectedKeys();
				Iterator<SelectionKey> iterator = selectKeys.iterator();

				while (iterator.hasNext()) {
					SelectionKey key = iterator.next();
					if ((key.readyOps() & SelectionKey.OP_ACCEPT) == SelectionKey.OP_ACCEPT) {
						ServerSocketChannel ssc1 = (ServerSocketChannel) key.channel();
						SocketChannel sc = ssc1.accept();
						sc.configureBlocking(false);
						sc.register(selector, SelectionKey.OP_READ);
						iterator.remove();
						System.out.println("Got connection from " + sc);
					} else if ((key.readyOps() & SelectionKey.OP_READ) == SelectionKey.OP_READ) {
						SocketChannel sc = (SocketChannel) key.channel();
						if (key.attachment() == null) {
							ByteBuffer buff = ByteBuffer.allocate(1024);
							key.attach(buff);
							System.out.println("first recv data from " + sc);
						}
						pool.execute(new IOWorker(sc, (ByteBuffer) key.attachment()));
						iterator.remove();
					}
				}
			}
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

}

class IOWorker implements Runnable {
	ByteBuffer buffer = null;
	SocketChannel sC = null;
	Charset gb2312 = Charset.forName("gb2312");
	CharsetDecoder decoder = gb2312.newDecoder();

	public IOWorker(SocketChannel sC, ByteBuffer buffer) {
		super();
		this.sC = sC;
		this.buffer = buffer;

	}

	@Override
	public void run() {
		// TODO Auto-generated method stub
		int r;
		try {
			r = this.sC.read(buffer);
			if (r > 0) {
				System.err.println("recv " + r + " new bytes from " + sC);
				buffer.flip();
				// get first byte
				// decode bytes stream
				for (; buffer.hasRemaining();) {
					byte fisrt = buffer.get(0);
					if (Byte.toUnsignedInt(fisrt) <= 127) {// 高字节的大小判断
						fisrt = buffer.get();
						System.err.println("[to " + sC.socket().getPort() + " ] "+ (char) fisrt);
					} else if (buffer.remaining() >= 2) {
						byte[] b = new byte[2];
						b[0] = buffer.get();
						b[1] = buffer.get();
						String s = new String(b, "gbk");
						System.err.println("[to " + sC.socket().getPort() + " ] " + s);
					} else
						break;
				}
				buffer.compact();
			}

		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

}

• （4）异步I/O

package iomodel;

import java.io.File;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousFileChannel;
import java.nio.channels.AsynchronousServerSocketChannel;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.StandardOpenOption;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
public class AIODemo {

	public static void main(String[] args) {
		// TODO Auto-generated method stub
		try {
			AsynchronousFileChannel aFC = AsynchronousFileChannel.open(Paths.get("a.txt"), StandardOpenOption.READ);
			System.out.println("main " + Thread.currentThread().getId());
			ByteBuffer buffer = ByteBuffer.allocate((int) new File("a.txt").length());
			Future<Integer> result = aFC.read(buffer, 0);	
			while (!result.isDone()) {
				System.out.println("i am doing other useful work...");
			}
			buffer.flip();
			System.out.println(result.get());
		
			System.out.println("read is done");
			
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (ExecutionException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
	}

}

class FileReadHandler implements CompletionHandler<Integer, AsynchronousFileChannel> {
	
	@Override
	public void completed(Integer result, AsynchronousFileChannel attachment) {
		System.out.println("new connection from " + result);
	}

	@Override
	public void failed(Throwable exc, AsynchronousFileChannel attachment) {
		// TODO Auto-generated method stub
		
	}
	
}

3. 总结

• 几种IO模型的优缺点，模型的阻塞、非阻塞以及同步、异步指的是什么？
• select模型与阻塞+多线程模型的区别以及使用场景；

4. References

[1] 史蒂文斯，芬纳，鲁道夫.UNIX网络编程卷一[M].北京:人民邮电出版社,2015:122-148